(ring.h). Move net and blk split drivers to use it.
blktap probably needs some work. :-)
Signed-off-by: Keir Fraser <keir@xensource.com>
blkif_response_t *resp;
unsigned long flags;
blkif_back_ring_t *blk_ring = &blkif->blk_ring;
+ int notify;
/* Place on the response ring for the relevant domain. */
spin_lock_irqsave(&blkif->blk_ring_lock, flags);
resp->id = id;
resp->operation = op;
resp->status = st;
- wmb(); /* Ensure other side can see the response fields. */
blk_ring->rsp_prod_pvt++;
- RING_PUSH_RESPONSES(blk_ring);
+ RING_PUSH_RESPONSES_AND_CHECK_NOTIFY(blk_ring, notify);
spin_unlock_irqrestore(&blkif->blk_ring_lock, flags);
- /* Kick the relevant domain. */
- notify_remote_via_irq(blkif->irq);
+ /*
+ * Tail check for pending requests. Allows frontend to avoid
+ * notifications if requests are already in flight (lower overheads
+ * and promotes batching).
+ */
+ if (!__on_blkdev_list(blkif) &&
+ RING_HAS_UNCONSUMED_REQUESTS(blk_ring)) {
+ add_to_blkdev_list_tail(blkif);
+ maybe_trigger_blkio_schedule();
+ }
+
+ if (notify)
+ notify_remote_via_irq(blkif->irq);
}
void blkif_deschedule(blkif_t *blkif)
static inline void flush_requests(struct blkfront_info *info)
{
+ RING_IDX old_prod = info->ring.sring->req_prod;
+
RING_PUSH_REQUESTS(&info->ring);
- notify_remote_via_irq(info->irq);
+
+ /*
+ * Send new requests /then/ check if any old requests are still in
+ * flight. If so then there is no need to send a notification.
+ */
+ mb();
+ if (info->ring.sring->rsp_prod == old_prod)
+ notify_remote_via_irq(info->irq);
}
static void kick_pending_request_queues(struct blkfront_info *info)
return IRQ_HANDLED;
}
+ again:
rp = info->ring.sring->rsp_prod;
rmb(); /* Ensure we see queued responses up to 'rp'. */
info->ring.rsp_cons = i;
+ if (i != info->ring.req_prod_pvt) {
+ int more_to_do;
+ RING_FINAL_CHECK_FOR_RESPONSES(&info->ring, more_to_do);
+ if (more_to_do)
+ goto again;
+ } else {
+ info->ring.sring->rsp_event = i + 1;
+ }
+
kick_pending_request_queues(info);
spin_unlock_irqrestore(&blkif_io_lock, flags);
kfree(copy);
- /* info->ring->req_prod will be set when we flush_requests().*/
- wmb();
-
/* Kicks things back into life. */
flush_requests(info);
static unsigned int blktap_poll(struct file *file, poll_table *wait)
{
poll_wait(file, &blktap_wait, wait);
- if (RING_HAS_UNPUSHED_REQUESTS(&blktap_ufe_ring)) {
+ if (blktap_ufe_ring.req_prod_pvt != blktap_ufe_ring.sring->req_prod) {
flush_tlb_all();
RING_PUSH_REQUESTS(&blktap_ufe_ring);
return POLLIN | POLLRDNORM;
static gnttab_transfer_t grant_rx_op[MAX_PENDING_REQS];
static unsigned char rx_notify[NR_IRQS];
-/* Don't currently gate addition of an interface to the tx scheduling list. */
-#define tx_work_exists(_if) (1)
-
static unsigned long mmap_vstart;
#define MMAP_VADDR(_req) (mmap_vstart + ((_req) * PAGE_SIZE))
* aggressive in avoiding new-packet notifications -- frontend only needs to
* send a notification if there are no outstanding unreceived responses.
* If we may be buffer transmit buffers for any reason then we must be rather
- * more conservative and advertise that we are 'sleeping' this connection here.
+ * more conservative and treat this as the final check for pending work.
*/
void netif_schedule_work(netif_t *netif)
{
- if (RING_HAS_UNCONSUMED_REQUESTS(&netif->tx)) {
+ int more_to_do;
+
+#ifdef CONFIG_XEN_NETDEV_PIPELINED_TRANSMITTER
+ more_to_do = RING_HAS_UNCONSUMED_REQUESTS(&netif->tx);
+#else
+ RING_FINAL_CHECK_FOR_REQUESTS(&netif->tx, more_to_do);
+#endif
+
+ if (more_to_do) {
add_to_net_schedule_list_tail(netif);
maybe_schedule_tx_action();
}
-#ifndef CONFIG_XEN_NETDEV_PIPELINED_TRANSMITTER
- else {
- netif->tx.sring->server_is_sleeping = 1;
- mb();
- if (RING_HAS_UNCONSUMED_REQUESTS(&netif->tx)) {
- netif->tx.sring->server_is_sleeping = 0;
- add_to_net_schedule_list_tail(netif);
- maybe_schedule_tx_action();
- }
- }
-#endif
}
void netif_deschedule_work(netif_t *netif)
pending_ring[MASK_PEND_IDX(pending_prod++)] = pending_idx;
- /*
- * Scheduling checks must happen after the above response is
- * posted. This avoids a possible race with a guest OS on
- * another CPU if that guest is testing against 'resp_prod'
- * when deciding whether to notify us when it queues additional
- * packets.
- */
- mb();
-
- if (RING_HAS_UNCONSUMED_REQUESTS(&netif->tx)) {
- add_to_net_schedule_list_tail(netif);
- } else {
- netif->tx.sring->server_is_sleeping = 1;
- mb();
- if (RING_HAS_UNCONSUMED_REQUESTS(&netif->tx)) {
- netif->tx.sring->server_is_sleeping = 0;
- add_to_net_schedule_list_tail(netif);
- }
- }
-
netif_put(netif);
}
}
RING_IDX i;
gnttab_map_grant_ref_t *mop;
unsigned int data_len;
- int ret;
+ int ret, work_to_do;
if (dealloc_cons != dealloc_prod)
net_tx_action_dealloc();
netif_get(netif);
remove_from_net_schedule_list(netif);
- /* Work to do? */
- if (!RING_HAS_UNCONSUMED_REQUESTS(&netif->tx)) {
+ RING_FINAL_CHECK_FOR_REQUESTS(&netif->tx, work_to_do);
+ if (!work_to_do) {
netif_put(netif);
continue;
}
irqreturn_t netif_be_int(int irq, void *dev_id, struct pt_regs *regs)
{
netif_t *netif = dev_id;
- if (tx_work_exists(netif)) {
- add_to_net_schedule_list_tail(netif);
- maybe_schedule_tx_action();
- }
+ add_to_net_schedule_list_tail(netif);
+ maybe_schedule_tx_action();
return IRQ_HANDLED;
}
{
RING_IDX i = netif->tx.rsp_prod_pvt;
netif_tx_response_t *resp;
+ int notify;
resp = RING_GET_RESPONSE(&netif->tx, i);
resp->id = id;
resp->status = st;
- wmb();
- netif->tx.rsp_prod_pvt = ++i;
- RING_PUSH_RESPONSES(&netif->tx);
- mb(); /* Update producer before checking event threshold. */
- if (i == netif->tx.sring->rsp_event)
+ netif->tx.rsp_prod_pvt = ++i;
+ RING_PUSH_RESPONSES_AND_CHECK_NOTIFY(&netif->tx, notify);
+ if (notify)
notify_remote_via_irq(netif->irq);
+
+#ifdef CONFIG_XEN_NETDEV_PIPELINED_TRANSMITTER
+ if (i == netif->tx.req_cons) {
+ int more_to_do;
+ RING_FINAL_CHECK_FOR_REQUESTS(&netif->tx, more_to_do);
+ if (more_to_do)
+ add_to_net_schedule_list_tail(netif);
+ }
+#endif
}
static int make_rx_response(netif_t *netif,
{
RING_IDX i = netif->rx.rsp_prod_pvt;
netif_rx_response_t *resp;
+ int notify;
resp = RING_GET_RESPONSE(&netif->rx, i);
resp->offset = offset;
resp->status = (s16)size;
if (st < 0)
resp->status = (s16)st;
- wmb();
+
netif->rx.rsp_prod_pvt = ++i;
- RING_PUSH_RESPONSES(&netif->rx);
+ RING_PUSH_RESPONSES_AND_CHECK_NOTIFY(&netif->rx, notify);
- mb(); /* Update producer before checking event threshold. */
- return (i == netif->rx.sring->rsp_event);
+ return notify;
}
static irqreturn_t netif_be_dbg(int irq, void *dev_id, struct pt_regs *regs)
RING_IDX i;
grant_ref_t ref;
unsigned long mfn;
+ int notify;
if (unlikely(np->tx_full)) {
printk(KERN_ALERT "%s: full queue wasn't stopped!\n",
tx->size = skb->len;
tx->csum_blank = (skb->ip_summed == CHECKSUM_HW);
- wmb(); /* Ensure that backend will see the request. */
np->tx.req_prod_pvt = i + 1;
- RING_PUSH_REQUESTS(&np->tx);
+ RING_PUSH_REQUESTS_AND_CHECK_NOTIFY(&np->tx, notify);
+ if (notify)
+ notify_remote_via_irq(np->irq);
network_tx_buf_gc(dev);
np->stats.tx_bytes += skb->len;
np->stats.tx_packets++;
- /* Only notify Xen if we really have to. */
- mb();
- if (np->tx.sring->server_is_sleeping) {
- np->tx.sring->server_is_sleeping = 0;
- notify_remote_via_irq(np->irq);
- }
-
return 0;
drop:
rx->id, rx->status);
RING_GET_REQUEST(&np->rx, np->rx.req_prod_pvt)->id =
rx->id;
- wmb();
+ RING_GET_REQUEST(&np->rx, np->rx.req_prod_pvt)->gref =
+ ref;
np->rx.req_prod_pvt++;
RING_PUSH_REQUESTS(&np->rx);
work_done--;
if (work_done < budget) {
local_irq_save(flags);
- np->rx.sring->rsp_event = i + 1;
-
- /* Deal with hypervisor racing our resetting of rx_event. */
- mb();
- if (np->rx.sring->rsp_prod == i) {
+ RING_FINAL_CHECK_FOR_RESPONSES(&np->rx, more_to_do);
+ if (!more_to_do)
__netif_rx_complete(dev);
- more_to_do = 0;
- }
local_irq_restore(flags);
}
/* Step 1: Reinitialise variables. */
np->tx_full = 0;
- np->rx.sring->rsp_event = np->tx.sring->rsp_event = 1;
/*
* Step 2: Rebuild the RX and TX ring contents.
np->stats.tx_bytes += skb->len;
np->stats.tx_packets++;
}
- wmb();
+
np->tx.req_prod_pvt = requeue_idx;
RING_PUSH_REQUESTS(&np->tx);
RING_GET_REQUEST(&np->rx, requeue_idx)->id = i;
requeue_idx++;
}
- wmb();
+
np->rx.req_prod_pvt = requeue_idx;
RING_PUSH_REQUESTS(&np->rx);
* packets.
*/
np->backend_state = BEST_CONNECTED;
- wmb();
notify_remote_via_irq(np->irq);
network_tx_buf_gc(dev);
#include "ring.h"
+/*
+ * Front->back notifications: When enqueuing a new request, there is no
+ * need to send a notification if there are old requests still in flight
+ * (that is, old_req_prod != sring->rsp_prod). The backend guarantees to check
+ * for new requests after queuing the response for the last in-flight request.
+ * (NB. The generic req_event mechanism is not used for blk requests.)
+ *
+ * Back->front notifications: When enqueuing a new response, sending a
+ * notification can be made conditional on rsp_event (i.e., the generic
+ * hold-off mechanism provided by the ring macros). Frontends must set
+ * rsp_event appropriately (e.g., using RING_FINAL_CHECK_FOR_RESPONSES()).
+ */
+
#ifndef blkif_vdev_t
#define blkif_vdev_t uint16_t
#endif
#include "ring.h"
+/*
+ * Note that there is *never* any need to notify the backend when enqueuing
+ * receive requests (netif_rx_request_t). Notifications after enqueuing any
+ * other type of message should be conditional on the appropriate req_event
+ * or rsp_event field in the shared ring.
+ */
+
typedef struct netif_tx_request {
grant_ref_t gref; /* Reference to buffer page */
uint16_t offset:15; /* Offset within buffer page */
-
-
-
-/*
+/******************************************************************************
+ * ring.h
+ *
* Shared producer-consumer ring macros.
+ *
* Tim Deegan and Andrew Warfield November 2004.
- */
+ */
#ifndef __XEN_PUBLIC_IO_RING_H__
#define __XEN_PUBLIC_IO_RING_H__
(__RD32(((_sz) - (long)&(_s)->ring + (long)(_s)) / sizeof((_s)->ring[0])))
/*
- * Macros to make the correct C datatypes for a new kind of ring.
+ * Macros to make the correct C datatypes for a new kind of ring.
*
- * To make a new ring datatype, you need to have two message structures,
- * let's say request_t, and response_t already defined.
+ * To make a new ring datatype, you need to have two message structures,
+ * let's say request_t, and response_t already defined.
*
- * In a header where you want the ring datatype declared, you then do:
+ * In a header where you want the ring datatype declared, you then do:
*
* DEFINE_RING_TYPES(mytag, request_t, response_t);
*
- * These expand out to give you a set of types, as you can see below.
- * The most important of these are:
+ * These expand out to give you a set of types, as you can see below.
+ * The most important of these are:
*
* mytag_sring_t - The shared ring.
* mytag_front_ring_t - The 'front' half of the ring.
* mytag_back_ring_t - The 'back' half of the ring.
*
- * To initialize a ring in your code you need to know the location and size
- * of the shared memory area (PAGE_SIZE, for instance). To initialise
- * the front half:
+ * To initialize a ring in your code you need to know the location and size
+ * of the shared memory area (PAGE_SIZE, for instance). To initialise
+ * the front half:
*
- * mytag_front_ring_t front_ring;
+ * mytag_front_ring_t front_ring;
+ * SHARED_RING_INIT((mytag_sring_t *)shared_page);
+ * FRONT_RING_INIT(&front_ring, (mytag_sring_t *)shared_page, PAGE_SIZE);
*
- * SHARED_RING_INIT((mytag_sring_t *)shared_page);
- * FRONT_RING_INIT(&front_ring, (mytag_sring_t *)shared_page, PAGE_SIZE);
+ * Initializing the back follows similarly (note that only the front
+ * initializes the shared ring):
*
- * Initializing the back follows similarly...
+ * mytag_back_ring_t back_ring;
+ * BACK_RING_INIT(&back_ring, (mytag_sring_t *)shared_page, PAGE_SIZE);
*/
#define DEFINE_RING_TYPES(__name, __req_t, __rsp_t) \
\
/* Shared ring page */ \
struct __name##_sring { \
- RING_IDX req_prod; \
- RING_IDX rsp_prod; \
- RING_IDX rsp_event; /* notify client when rsp_prod == rsp_event */ \
- uint8_t server_is_sleeping; /* notify server to kick off work */ \
+ RING_IDX req_prod, req_event; \
+ RING_IDX rsp_prod, rsp_event; \
union __name##_sring_entry ring[1]; /* variable-length */ \
}; \
\
typedef struct __name##_back_ring __name##_back_ring_t
/*
- * Macros for manipulating rings.
+ * Macros for manipulating rings.
*
- * FRONT_RING_whatever works on the "front end" of a ring: here
- * requests are pushed on to the ring and responses taken off it.
+ * FRONT_RING_whatever works on the "front end" of a ring: here
+ * requests are pushed on to the ring and responses taken off it.
*
- * BACK_RING_whatever works on the "back end" of a ring: here
- * requests are taken off the ring and responses put on.
+ * BACK_RING_whatever works on the "back end" of a ring: here
+ * requests are taken off the ring and responses put on.
*
- * N.B. these macros do NO INTERLOCKS OR FLOW CONTROL.
- * This is OK in 1-for-1 request-response situations where the
- * requestor (front end) never has more than RING_SIZE()-1
- * outstanding requests.
+ * N.B. these macros do NO INTERLOCKS OR FLOW CONTROL.
+ * This is OK in 1-for-1 request-response situations where the
+ * requestor (front end) never has more than RING_SIZE()-1
+ * outstanding requests.
*/
/* Initialising empty rings */
#define SHARED_RING_INIT(_s) do { \
- (_s)->req_prod = 0; \
- (_s)->rsp_prod = 0; \
+ (_s)->req_prod = (_s)->rsp_prod = 0; \
+ (_s)->req_event = (_s)->rsp_event = 1; \
} while(0)
#define FRONT_RING_INIT(_r, _s, __size) do { \
#define RING_SIZE(_r) \
((_r)->nr_ents)
-/* How many empty slots are on a ring? */
-#define RING_PENDING_REQUESTS(_r) \
- ( ((_r)->req_prod_pvt - (_r)->rsp_cons) )
-
/* Test if there is an empty slot available on the front ring.
* (This is only meaningful from the front. )
*/
(((_r)->req_cons - (_r)->rsp_prod_pvt) != \
RING_SIZE(_r)) )
-/* Test if there are messages waiting to be pushed. */
-#define RING_HAS_UNPUSHED_REQUESTS(_r) \
- ( (_r)->req_prod_pvt != (_r)->sring->req_prod )
-
-#define RING_HAS_UNPUSHED_RESPONSES(_r) \
- ( (_r)->rsp_prod_pvt != (_r)->sring->rsp_prod )
-
-/* Copy the private producer pointer into the shared ring so the other end
- * can see the updates we've made. */
-#define RING_PUSH_REQUESTS(_r) do { \
- wmb(); \
- (_r)->sring->req_prod = (_r)->req_prod_pvt; \
-} while (0)
-
-#define RING_PUSH_RESPONSES(_r) do { \
- wmb(); \
- (_r)->sring->rsp_prod = (_r)->rsp_prod_pvt; \
-} while (0)
-
/* Direct access to individual ring elements, by index. */
#define RING_GET_REQUEST(_r, _idx) \
(&((_r)->sring->ring[ \
#define RING_REQUEST_CONS_OVERFLOW(_r, _cons) \
(((_cons) - (_r)->rsp_prod_pvt) >= RING_SIZE(_r))
+#define RING_PUSH_REQUESTS(_r) do { \
+ wmb(); /* back sees requests /before/ updated producer index */ \
+ (_r)->sring->req_prod = (_r)->req_prod_pvt; \
+} while (0)
+
+#define RING_PUSH_RESPONSES(_r) do { \
+ wmb(); /* front sees responses /before/ updated producer index */ \
+ (_r)->sring->rsp_prod = (_r)->rsp_prod_pvt; \
+} while (0)
+
+/*
+ * Notification hold-off (req_event and rsp_event):
+ *
+ * When queueing requests or responses on a shared ring, it may not always be
+ * necessary to notify the remote end. For example, if requests are in flight
+ * in a backend, the front may be able to queue further requests without
+ * notifying the back (if the back checks for new requests when it queues
+ * responses).
+ *
+ * When enqueuing requests or responses:
+ *
+ * Use RING_PUSH_{REQUESTS,RESPONSES}_AND_CHECK_NOTIFY(). The second argument
+ * is a boolean return value. True indicates that the receiver requires an
+ * asynchronous notification.
+ *
+ * After dequeuing requests or responses (before sleeping the connection):
+ *
+ * Use RING_FINAL_CHECK_FOR_REQUESTS() or RING_FINAL_CHECK_FOR_RESPONSES().
+ * The second argument is a boolean return value. True indicates that there
+ * are pending messages on the ring (i.e., the connection should not be put
+ * to sleep).
+ *
+ * These macros will set the req_event/rsp_event field to trigger a
+ * notification on the very next message that is enqueued. If you want to
+ * create batches of work (i.e., only receive a notification after several
+ * messages have been enqueued) then you will need to create a customised
+ * version of the FINAL_CHECK macro in your own code, which sets the event
+ * field appropriately.
+ */
+
+#define RING_PUSH_REQUESTS_AND_CHECK_NOTIFY(_r, _notify) do { \
+ RING_IDX __old = (_r)->sring->req_prod; \
+ RING_IDX __new = (_r)->req_prod_pvt; \
+ wmb(); /* back sees requests /before/ updated producer index */ \
+ (_r)->sring->req_prod = __new; \
+ mb(); /* back sees new requests /before/ we check req_event */ \
+ (_notify) = ((RING_IDX)(__new - (_r)->sring->req_event) < \
+ (RING_IDX)(__new - __old)); \
+} while (0)
+
+#define RING_PUSH_RESPONSES_AND_CHECK_NOTIFY(_r, _notify) do { \
+ RING_IDX __old = (_r)->sring->rsp_prod; \
+ RING_IDX __new = (_r)->rsp_prod_pvt; \
+ wmb(); /* front sees responses /before/ updated producer index */ \
+ (_r)->sring->rsp_prod = __new; \
+ mb(); /* front sees new responses /before/ we check rsp_event */ \
+ (_notify) = ((RING_IDX)(__new - (_r)->sring->rsp_event) < \
+ (RING_IDX)(__new - __old)); \
+} while (0)
+
+#define RING_FINAL_CHECK_FOR_REQUESTS(_r, _work_to_do) do { \
+ (_work_to_do) = RING_HAS_UNCONSUMED_REQUESTS(_r); \
+ if (_work_to_do) break; \
+ (_r)->sring->req_event = (_r)->req_cons + 1; \
+ mb(); \
+ (_work_to_do) = RING_HAS_UNCONSUMED_REQUESTS(_r); \
+} while (0)
+
+#define RING_FINAL_CHECK_FOR_RESPONSES(_r, _work_to_do) do { \
+ (_work_to_do) = RING_HAS_UNCONSUMED_RESPONSES(_r); \
+ if (_work_to_do) break; \
+ (_r)->sring->rsp_event = (_r)->rsp_cons + 1; \
+ mb(); \
+ (_work_to_do) = RING_HAS_UNCONSUMED_RESPONSES(_r); \
+} while (0)
+
#endif /* __XEN_PUBLIC_IO_RING_H__ */
/*
projects / xen.git / commitdiff